package com.feiwei.udf

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.DataTypes
import org.apache.flink.table.api.scala.StreamTableEnvironment
import org.apache.flink.table.descriptors.{Csv, FileSystem, Schema}
import org.apache.flink.table.functions.ScalarFunction
import org.apache.flink.table.api._
import org.apache.flink.table.api.scala._
import org.apache.flink.types.Row

import org.apache.flink.streaming.api._

import org.apache.flink.streaming.api._
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.DataTypes
import org.apache.flink.table.api.scala._
import org.apache.flink.table.descriptors._
import org.apache.flink.streaming.api.scala._

object day12_UDFTest {

  def main(args: Array[String]): Unit = {

    val environment = StreamExecutionEnvironment.getExecutionEnvironment


    val tableEnv = StreamTableEnvironment.create(environment)


    val filePath = "E:\\repository\\company\\myself\\flink-learning\\flink-learning-demo\\src\\main\\resources\\sensor.txt"

    tableEnv.connect(new FileSystem().path(filePath))
      .withFormat(new Csv())
      .withSchema(new Schema()
        .field("id", DataTypes.STRING())
        .field("timestamp", DataTypes.BIGINT())
        .field("temp", DataTypes.DOUBLE())
      )
      .createTemporaryTable("inputTable")

    //2转换操作
    val source = tableEnv.from("inputTable")

    //3 新建一个udf函数

    val hashCode = new HashCode(23)

    //4.1 api
    val t1 = source.select('id, 'timestamp, hashCode('id))

    //4.2 sql

    tableEnv.createTemporaryView("sensor", source)

    tableEnv.registerFunction("hashCode", hashCode)

    //timestamp  sql里面是关键字
    val t2 = tableEnv.sqlQuery(
      "select id,temp,`timestamp` as ts,hashCode(id) from sensor"
    )

    t1.toAppendStream[Row].print("api")

    t2.toAppendStream[Row].print("sql")

    environment.execute()

  }


}

/**
  * 自定义的UDF函数
  *
  * @param f
  */
class HashCode(f: Int) extends ScalarFunction {

  def eval(s: String): Int = {
    s.hashCode * f - 1000
  }
}

